/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.scheduler;

import java.io.IOException;

import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.DeploymentException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.exchange.ConditionFlag;
import cgl.granules.exchange.DeploymentMessage;
import cgl.granules.exchange.ExchangeFactory;
import cgl.granules.exchange.ExchangeType;
import cgl.granules.exchange.ExecutionStatus;
import cgl.granules.exchange.ResourceDiscovery;
import cgl.granules.operation.Operation;
import cgl.granules.operation.ProcessingException;
import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.util.UUIDRetriever;

/**
 * This is a class which is reponsible for processing various exchanges related
 * to the deployment, discovery and advertisement of application instances and
 * resources.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class ExchangeProcessor implements NBEventListener {
	private StreamingService streamingService;
	private EventConsumer consumer;
	private EventProducer producer;
	private UUIDRetriever retriever;
	private ExchangeFactory exchangeFactory;
	private String deploymentStream;
	private String discoveryStream = "Granules/Resources/Discovery";
	private String moduleName = "ExchangeProcessor: ";
	private TaskManagement taskManager;
	private Diagnostics diagnostics;
	private boolean debug = false;

	public ExchangeProcessor(Diagnostics diagnostics, String deploymentStream,
			int numOfThreads) {
		this.diagnostics = diagnostics;
		this.deploymentStream = deploymentStream;
		streamingService = StreamingService.getInstance();
		exchangeFactory = ExchangeFactory.getInstance();
		retriever = UUIDRetriever.getInstance();
		taskManager = new TaskManagement(numOfThreads, diagnostics);
	}

	public void initialize() throws CommunicationsException {
		consumer = streamingService.createConsumer(this);
		producer = streamingService.createProducer();
		consumeStream(deploymentStream, StreamingAccess.STRING_SYNOPSIS);
		consumeStream(discoveryStream, StreamingAccess.STRING_SYNOPSIS);
	}

	@Override
	public void onEvent(NBEvent nbEvent) {
		/*
		 * Also do a check to make sure that you have received something that
		 * was sent on an authorized stream.
		 */

		byte[] exchangeBytes = nbEvent.getContentPayload();

		try {
			processExchange(exchangeBytes);
		} catch (SuppressedException e) {
			System.out.println(e.getMessage());
		}
	}

	/**
	 * Process the exchange, and then fork processing to appropriate derived
	 * classes of the Exchange
	 * 
	 * @param exchangeBytes
	 * @throws SuppressedException
	 */
	private void processExchange(byte[] exchangeBytes)
			throws SuppressedException {
		String stage = "Investigating content payload";
		String problem = null;

		if (exchangeBytes == null) {
			problem = "The content payload for the exchange is NULL";
			throwSupressedException(stage, problem);
		}

		stage = "Checking validity of the exchange type";
		int exchangeType = exchangeFactory.getExchangeType(exchangeBytes);
		if (!exchangeFactory.isValidExchangeType(exchangeType)) {
			problem = "The encapsulated exchange type = " + exchangeType
					+ " is an invalid one.";
			throwSupressedException(stage, problem);
		}

		try {
			if (exchangeType == ExchangeType.DEPLOYMENT) {
				stage = "Reconstructing the DeploymentMessage ";
				DeploymentMessage deploymentRequest = exchangeFactory
						.getDeploymentMessage(exchangeBytes);
				processDeploymentMessage(deploymentRequest);
			}

			if (exchangeType == ExchangeType.EXECUTION_STATUS) {
				stage = "Reconstructing the ExecutionStatus exchange ";
				ExecutionStatus executionStatus = exchangeFactory
						.getExecutionStatus(exchangeBytes);
				diagnostics.processExecutionStatusRequest(executionStatus);
			}

			if (exchangeType == ExchangeType.RESOURCE_DISCOVERY) {
				ResourceDiscovery resourceDiscovery = exchangeFactory
						.getResourceDiscovery(exchangeBytes);
				diagnostics.processResourceDiscoveryRequest(resourceDiscovery);
			}

		} catch (IOException e) {
			problem = " there was an IOException " + e.getMessage();
			throwSupressedException(stage, problem);
		} catch (MarshallingException e) {
			problem = " there was a MarshallingException " + e.getMessage();
			throwSupressedException(stage, problem);
		}
	}

	/**
	 * Reconstruct and process the deployment message that has been received.
	 * 
	 * @param exchangeBytes
	 * @throws SuppressedException
	 */

	private void processDeploymentMessage(DeploymentMessage deploymentRequest)
			throws SuppressedException {
		String stage = null;
		String problem = null;

		stage = "Inspecting the DeploymentMessage ";
		String jobIdentifier = deploymentRequest.getJobIdentifier();
		int exchangeSubType = deploymentRequest.getExchangeSubType();
		/* Check to see if the deployment message was previously received */
		boolean wasPreviouslyProcessed = false;
		if (wasPreviouslyProcessed) {
			return;
		}

		stage = "Create & Send a Response to an invalid DeploymentMessage";
		if (jobIdentifier == null) {
			problem = "An invalid JobIdentifier=NULL has been specified.";
			sendDeploymentResponse(deploymentRequest, "Unassigned",
					ConditionFlag.PROBLEMS_ENCOUNTERED, problem, stage);
		}

		if (!deploymentRequest.isValidExchangeSubType()) {
			problem = "The exchangeSubType = " + exchangeSubType
					+ " is invalid";
			sendDeploymentResponse(deploymentRequest, "Unassigned",
					ConditionFlag.PROBLEMS_ENCOUNTERED, problem, stage);
		}

		/* Check for problems with the Response indicator */
		if (deploymentRequest.isResponse()) {
			problem = "A Deployment RESPONSE was incorrectly received";
			sendDeploymentResponse(deploymentRequest, "Unassigned",
					ConditionFlag.PROBLEMS_ENCOUNTERED, problem, stage);
		} /* deploymentRequest.isResponse() */

		if (exchangeSubType == DeploymentMessage.DEPLOY) {
			deployOperation(deploymentRequest);
		}

		if (exchangeSubType == DeploymentMessage.STATUS) {
			getStatusOfTask(deploymentRequest);
		}

	}

	/**
	 * Deploy the operation
	 * 
	 * @param deploymentRequest
	 * @throws SuppressedException
	 */
	private void deployOperation(DeploymentMessage deploymentRequest)
			throws SuppressedException {

		String stage = "Deploying the Operation ";
		String problem = "";
		if (debug) {
			System.out.println(stage);
		}
		String taskIdentifier = retriever.getRandomBasedUUIDAsString();
		try {
			Operation operation = deploymentRequest.getOperation();
			diagnostics.addDeploymentRequest(taskIdentifier, deploymentRequest);
			taskManager.manageExecution(taskIdentifier, operation);
			String description = "Successfully DEPLOYED the operation";
			// sendDeploymentResponse(deploymentRequest, taskIdentifier,
			// ConditionFlag.SUCCESS, description, stage);
			return;
		} catch (ProcessingException e) {
			problem = "Encountered a ProcessingException during Deployment: "
					+ e.getMessage();
		} catch (DatasetException e) {
			problem = "Encountered a DatasetException during Deployment: "
					+ e.getMessage();
		} catch (SchedulingException e) {
			problem = "Encountered a SchedulingException during Deployment: "
					+ e.getMessage();
		} catch (NullPointerException npe) {
			problem = npe.toString();
			npe.printStackTrace();

		}

		sendDeploymentResponse(deploymentRequest, taskIdentifier,
				ConditionFlag.FAILURE, problem, stage);
	}

	/**
	 * Process deployment message that is requesting the status of a previously
	 * submitted task.
	 * 
	 * @param deploymentRequest
	 *            The issued
	 * @throws SuppressedException
	 */
	private void getStatusOfTask(DeploymentMessage deploymentRequest)
			throws SuppressedException {
		String stage = "Retrieving the Status of Task ";
		String description = "";
		System.out.println(stage);
		int conditionFlag;
		String taskIdentifier = deploymentRequest.getTaskIdentifier();
		TaskLifecycleMetrics metrics = diagnostics
				.getStatusOfTask(taskIdentifier);

		if (metrics != null) {
			description = "Status of Task (" + taskIdentifier + ") "
					+ metrics.toString();
			conditionFlag = ConditionFlag.SUCCESS;
		} else {
			description = "Unable to locate task with ID = (" + taskIdentifier
					+ ").";
			conditionFlag = ConditionFlag.FAILURE;
		}

		sendDeploymentResponse(deploymentRequest, taskIdentifier,
				conditionFlag, description, stage);
	}

	/**
	 * This method creates and send a DeploymentMessage (RESPONSE)
	 * 
	 * @param deploymentRequest
	 *            The original request that was received
	 * @param taskIdentifier
	 *            The taskIdentifier that was assigned by the system
	 * 
	 * @param conditionFlag
	 *            The ConditionFlag that would be set in the response
	 * @param description
	 *            The description associated with the response
	 * @param stage
	 *            The stage at which this method was invoke. Useful for
	 *            SuppressedException block
	 * @throws SuppressedException
	 *             If there are problems that cannot be handled by another
	 *             component or if it cannot be sent somewhere else.
	 */
	private void sendDeploymentResponse(DeploymentMessage deploymentRequest,
			String taskIdentifier, int conditionFlag, String description,
			String stage) throws SuppressedException {

		String jobIdentifier = deploymentRequest.getJobIdentifier();
		int exchangeSubType = deploymentRequest.getExchangeSubType();
		String respondTo = deploymentRequest.getRespondTo();
		String problem = description;

		if (respondTo == null) {
			problem += ", and Without the [respondTo] field";
			throwSupressedException(stage, problem);
		}

		try {
			DeploymentMessage deploymentResponse = exchangeFactory
					.createDeploymentResponse(jobIdentifier, taskIdentifier,
							exchangeSubType, conditionFlag, problem);
			byte[] responseBytes = deploymentResponse.getBytes();
			publishToStream(respondTo, responseBytes);
		} catch (DeploymentException e) {
			problem = " there was a DeploymentException " + e.getMessage();
			throwSupressedException(stage, problem);
		} catch (IOException e) {
			problem = " there was an IOException " + e.getMessage();
			throwSupressedException(stage, problem);
		} catch (MarshallingException e) {
			problem = " there was a MarshallingException " + e.getMessage();
			throwSupressedException(stage, problem);
		} catch (CommunicationsException e) {
			problem = " there was a CommunicationsException " + e.getMessage();
			throwSupressedException(stage, problem);
		}

	}

	/**
	 * Utility method which allows a consumer to consume a specific stream
	 * 
	 * @param synopsisType
	 *            The synopsis type of the underlying stream
	 * @param streamSynopsis
	 *            The synopsis for the stream
	 * @throws CommunicationsException
	 *             When communication problems are encountered.
	 */
	private void consumeStream(String streamSynopsis, int synopsisType)
			throws CommunicationsException {
		try {
			Profile profile = streamingService.createProfile(streamSynopsis,
					synopsisType);
			consumer.subscribeTo(profile);
		} catch (ServiceException serEx) {
			throw new CommunicationsException(serEx.toString());
		}
	}

	/**
	 * Utility method that allows a producer to publish a set of bytes onto the
	 * specified stream.
	 * 
	 * @param publishStream
	 *            Synopsis of the stream to publish to
	 * @param transferBytes
	 *            the payload for the stream fragments
	 * @throws CommunicationsException
	 *             When communication problems are encountered.
	 */
	private void publishToStream(String publishStream, byte[] transferBytes)
			throws CommunicationsException {
		NBEvent nbEvent;
		try {
			nbEvent = producer.generateEvent(
					TemplateProfileAndSynopsisTypes.STRING, publishStream,
					transferBytes);
			producer.publishEvent(nbEvent);
		} catch (ServiceException e) {
			throwCommunicationsException(e.toString());
		}
	}

	private void reportErrorsThatCannotBePropagated(String errorReport) {
		String completeError = moduleName
				+ "This is an error that CANNOT be propagated." + errorReport
				+ " No further processing will be done";
		System.out.println(completeError);
	}

	/**
	 * Throws a MarshallingException with the specified reason
	 * 
	 * @param errorReport
	 * @throws MarshallingException
	 */
	private void throwMarshallingException(String errorReport)
			throws MarshallingException {
		throw new MarshallingException(errorReport);
	}

	/**
	 * Throws a Deployment Exception with the specified reason
	 * 
	 * @param reason
	 * @throws DeploymentException
	 */
	private void throwDeploymentException(String reason)
			throws DeploymentException {
		throw new DeploymentException(reason);
	}

	/**
	 * Throws a CommunicationsException with the specified reason
	 * 
	 * @param reason
	 * @throws CommunicationsException
	 */
	private void throwCommunicationsException(String reason)
			throws CommunicationsException {
		throw new CommunicationsException(reason);
	}

	private void throwSupressedException(String stage, String problem)
			throws SuppressedException {
		String errorReport = "Problems encountered when (" + stage
				+ ") because :-> " + problem;
		throw new SuppressedException(errorReport);
	}
}
